package cn.linjianhui.flink.sample;

import cn.linjianhui.flink.sample.model.OrderForChannel;
import cn.linjianhui.flink.sample.model.SaleChannelAccumulator;
import cn.linjianhui.flink.sample.util.ParamUtils;
import cn.linjianhui.flink.sample.util.Utils;
import com.alibaba.fastjson.JSON;
import cn.linjianhui.flink.sample.model.*;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;

import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.ContinuousProcessingTimeTrigger;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;

import java.math.BigDecimal;
import java.time.LocalDate;
import java.time.ZoneOffset;
import java.util.Properties;

public class SaleChannelJob {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = Utils.getStreamExecutionEnvironment();

        Properties omsProps = ParamUtils.getKafkaConsumerProperty(args, "oms-sale-channel", "earliest");
        DataStream<String> omsSourceStream = env.addSource(new FlinkKafkaConsumer011<>("OMS-Order", new SimpleStringSchema(), omsProps));

        DataStream<OrderForChannel> omsSaleChannelStream = omsSourceStream
                .map((MapFunction<String, OrderForChannel>) s -> JSON.parseObject(s, OrderForChannel.class))
                .filter((FilterFunction<OrderForChannel>) o -> {
                    long currentDayTimestamp = LocalDate.now().atStartOfDay(ZoneOffset.ofHours(8)).toInstant().toEpochMilli();
                    return o.getORDERSTATUS() == 3 && o.getPAYMENTDATE() >= currentDayTimestamp;
                });

        Properties wsProps = ParamUtils.getKafkaConsumerProperty(args, "ws-sale-channel", "earliest");
        DataStream<String> wsSourceStream = env.addSource(new FlinkKafkaConsumer011<>("WS-Order", new SimpleStringSchema(), wsProps));

        DataStream<OrderForChannel> wsSaleChannelStream = wsSourceStream
                .map((MapFunction<String, OrderForChannel>) s -> JSON.parseObject(s, OrderForChannel.class))
                .filter((FilterFunction<OrderForChannel>) o -> {
                    long currentDayTimestamp = LocalDate.now().atStartOfDay(ZoneOffset.ofHours(8)).toInstant().toEpochMilli();
                    return o.getORDERSTATUS() == 3 && o.getPAYMENTDATE() >= currentDayTimestamp;
                });

        DataStream<OrderForChannel> saleChannelStream = omsSaleChannelStream.union(wsSaleChannelStream);

        WindowedStream<OrderForChannel, Tuple, TimeWindow> siteDayWindowStream = saleChannelStream
                .keyBy("SALECHANNEL")
                .window(TumblingProcessingTimeWindows.of(Time.days(1), Time.hours(-8)))
                .trigger(ContinuousProcessingTimeTrigger.of(Time.seconds(1)));

        DataStream<SaleChannelAccumulator> saleChannelAggStream = siteDayWindowStream.aggregate(new SaleChannelAggregateFunc());

        DataStream<Tuple2<String, BigDecimal>> dataStream = saleChannelAggStream.map((MapFunction<SaleChannelAccumulator, Tuple2<String, BigDecimal>>) ac ->
                new Tuple2<>(ac.getSaleChannel(), ac.getPayAmount()))
                .returns(TypeInformation.of(new TypeHint<Tuple2<String, BigDecimal>>() {
                }));
        FlinkJedisPoolConfig jedisPoolConfig = ParamUtils.getJedisPoolConfig(args);
        dataStream.addSink(new RedisSink<>(jedisPoolConfig, new SaleChannelRedisMapper()));

        env.execute("SaleChannel analysis Job");

    }

    public final static class SaleChannelAggregateFunc implements AggregateFunction<OrderForChannel, SaleChannelAccumulator, SaleChannelAccumulator> {
        @Override
        public SaleChannelAccumulator createAccumulator() {
            return new SaleChannelAccumulator();
        }

        @Override
        public SaleChannelAccumulator add(OrderForChannel order, SaleChannelAccumulator accumulator) {
            if ("".equals(accumulator.getSaleChannel())) {
                accumulator.setSaleChannel(order.getSALECHANNEL());
            }
            if (order.getORDERSTATUS() == 3) {
                if (!accumulator.containsOrder(order.getORDERCODE())) {
                    accumulator.addPayAmount(new BigDecimal(order.getNETAMT()));
                    accumulator.addOrderId(order.getORDERCODE());
                }
            }
            return accumulator;
        }

        @Override
        public SaleChannelAccumulator getResult(SaleChannelAccumulator result) {
            return result;
        }

        @Override
        public SaleChannelAccumulator merge(SaleChannelAccumulator acc1, SaleChannelAccumulator acc2) {
            if ("".equals(acc1.getSaleChannel())) {
                acc1.setSaleChannel(acc2.getSaleChannel());
            }
            acc1.addPayAmount(acc2.getPayAmount());
            acc1.addOrderIds(acc2.getOrderIds());
            return acc1;
        }

    }

    public static final class SaleChannelRedisMapper implements RedisMapper<Tuple2<String, BigDecimal>> {
        private static final long serialVersionUID = 1L;
        private static final String HASH_NAME_PREFIX = "flink-sample-sale_channel";

        @Override
        public RedisCommandDescription getCommandDescription() {
            return new RedisCommandDescription(RedisCommand.ZADD, HASH_NAME_PREFIX);
        }

        @Override
        public String getKeyFromData(Tuple2<String, BigDecimal> tuple2) {
            return tuple2.f0;
        }

        @Override
        public String getValueFromData(Tuple2<String, BigDecimal> tuple2) {
            return String.valueOf(tuple2.f1);
        }

    }

}
